package simpledb;

import java.util.*;

/**
 * Knows how to compute some aggregate over a set of StringFields.
 */
public class StringAggregator implements Aggregator {

    private static final long serialVersionUID = 1L;

    private int groupByField;
    private Type groupByFieldType;
    private int aggregateField;
    private Aggregator.Op op;

    private HashMap hashMap=new HashMap();
    private TupleDesc tupleDesc;
    private Set set;
    private Iterator iterator;

    /**
     * Aggregate constructor
     * @param gbfield the 0-based index of the group-by field in the tuple, or NO_GROUPING if there is no grouping
     * @param gbfieldtype the type of the group by field (e.g., Type.INT_TYPE), or null if there is no grouping
     * @param afield the 0-based index of the aggregate field in the tuple
     * @param what aggregation operator to use -- only supports COUNT
     * @throws IllegalArgumentException if what != COUNT
     */

    public StringAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.groupByField=gbfield;
        this.groupByFieldType=gbfieldtype;
        this.aggregateField=afield;
        this.op=what;
        if (what!= Op.COUNT){
            throw new IllegalArgumentException("what != COUNT");
        }
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the constructor
     * @param tup the Tuple containing an aggregate field and a group-by field
     */
    @Override
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        if (groupByField==Aggregator.NO_GROUPING){
            if (hashMap.isEmpty()){
                Type[] types={Type.INT_TYPE};
                tupleDesc = new TupleDesc(types);
                hashMap.put(Aggregator.NO_GROUPING,1);
                return;
            }
            return;
        }
        Field key = tup.getField(groupByField);
        if (hashMap.isEmpty()) {
            Type[] types = {tup.getTupleDesc().getFieldType(groupByField),
                    Type.INT_TYPE};
            tupleDesc = new TupleDesc(types);
        }
        if (!hashMap.containsKey(key)){
            hashMap.put(key,1);
            return;
        }
        int value= (int) hashMap.get(key);
        value++;
        hashMap.remove(key);
        hashMap.put(key,value);
    }

    /**
     * Create a OpIterator over group aggregate results.
     *
     * @return a OpIterator whose tuples are the pair (groupVal,
     *   aggregateVal) if using group, or a single (aggregateVal) if no
     *   grouping. The aggregateVal is determined by the type of
     *   aggregate specified in the constructor.
     */
    @Override
    public OpIterator iterator() {
        // some code goes here
        return new OpIterator() {
            @Override
            public void open() throws DbException, TransactionAbortedException {
                set=hashMap.entrySet();
                iterator=set.iterator();
            }

            @Override
            public boolean hasNext() throws DbException, TransactionAbortedException {
                return iterator.hasNext();
            }

            @Override
            public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
                Map.Entry map= (Map.Entry) iterator.next();
                Tuple tuple=new Tuple(tupleDesc);
                if (groupByField==Aggregator.NO_GROUPING){
                    tuple.setField(0,new IntField((Integer) map.getValue()));
                }else {
                    tuple.setField(0, (Field) map.getKey());
                    tuple.setField(1, new IntField((Integer) map.getValue()));
                }
                return tuple;
            }

            @Override
            public void rewind() throws DbException, TransactionAbortedException {
                this.close();
                this.open();
            }

            @Override
            public TupleDesc getTupleDesc() {
                return tupleDesc;
            }

            @Override
            public void close() {
                iterator=null;
            }
        };
    }

}
